package com.lpssfxy.online.test

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{DefaultJedisClientConfig, Jedis}

// 定义配置对象
case class AppConfig(
                      sparkCores: String,
                      mongoUri: String,
                      mongoDb: String,
                      kafkaTopic: String,
                      kafkaBootstrapServers: String,
                      redisHost: String,
                      redisPort: Int,
                      redisPassword: String
                    )

// 定义标准推荐对象
case class Recommendation(productId: Int, score: Double)

// 定义用户的推荐列表
case class UserRecs(userId: Int, recs: Seq[Recommendation])

// 定义商品相似度列表
case class ProductRecs(productId: Int, recs: Seq[Recommendation])

object RealtimeRecommender {
  // 定义常量和表名
  private val MONGODB_RATING_COLLECTION = "rating"
  private val STREAM_RECS = "streamRecs"
  private val PRODUCT_RECS = "productRecs"

  private val MAX_USER_RATING_NUM = 20
  private val MAX_SIM_PRODUCTS_NUM = 20

  def main(args: Array[String]): Unit = {
    val config = AppConfig(
      sparkCores = "local[*]",
      mongoUri = "mongodb://fooadmin:123456@s3:27017/bigdata",
      mongoDb = "bigdata",
      kafkaTopic = "recommender",
      kafkaBootstrapServers = "s1:9092,s2:9092,s3:9092",
      redisHost = "192.168.85.144",
      redisPort = 6379,
      redisPassword = "openGauss_1234"
    )

    val (spark, sc, ssc) = createSparkContext(config)
    implicit val mongoConfig = MongoConfig(config.mongoUri, config.mongoDb)

    val simProcutsMatrixBC = loadAndBroadcastSimProductsMatrix(spark, mongoConfig)
    val ratingStream = createRatingStream(ssc, config)

    processRatingStream(ratingStream, simProcutsMatrixBC, config, mongoConfig)

    // 启动streaming
    ssc.start()
    println("streaming started!")
    ssc.awaitTermination()
  }

  private def createSparkContext(config: AppConfig): (SparkSession, SparkContext, StreamingContext) = {
    val sparkConf = new SparkConf().setMaster(config.sparkCores).setAppName("RealtimeRecommender")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(2))
    (spark, sc, ssc)
  }

  private def loadAndBroadcastSimProductsMatrix(spark: SparkSession, mongoConfig: MongoConfig) = {
    import spark.implicits._
    val simProductsMatrix = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", PRODUCT_RECS)
      .format("com.mongodb.spark.sql")
      .load()
      .as[ProductRecs]
      .rdd
      .map { item =>
        (item.productId, item.recs.map(x => (x.productId, x.score)).toMap)
      }
      .collectAsMap()
    spark.sparkContext.broadcast(simProductsMatrix)
  }

  private def createRatingStream(ssc: StreamingContext, config: AppConfig) = {
    val kafkaParam = Map(
      "bootstrap.servers" -> config.kafkaBootstrapServers,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest"
    )
    val kafkaStream = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array(config.kafkaTopic), kafkaParam)
    )
    kafkaStream.map { msg =>
      val attr = msg.value().split("\\|")
      (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt)
    }
  }

  private def processRatingStream(ratingStream: DStream[(Int, Int, Double, Int)],
                                  simProcutsMatrixBC: org.apache.spark.broadcast.Broadcast[scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]],
                                  config: AppConfig,
                                  mongoConfig: MongoConfig): Unit = {
    ratingStream.foreachRDD { rdds =>
      rdds.foreach {
        case (userId, productId, score, timestamp) =>
          println("rating data coming!>>>>>>>>>>>>>>>>>>")

          // 在每个节点上创建Jedis和MongoClient实例
          val clientConfig = DefaultJedisClientConfig.builder()
            .password(config.redisPassword)
            .build()
          val jedis = new Jedis(config.redisHost, config.redisPort, clientConfig)
          val mongoClient = MongoClient(MongoClientURI(config.mongoUri))
          try {
            // 1. 从redis里取出当前用户的最近评分，保存成一个数组Array[(productId, score)]
            val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATING_NUM, userId, jedis)

            // 2. 从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表，保存成一个数组Array[productId]
            val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId, userId, simProcutsMatrixBC.value)(mongoConfig, mongoClient)

            // 3. 计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表，保存成 Array[(productId, score)]
            val streamRecs = computeProductScore(candidateProducts, userRecentlyRatings, simProcutsMatrixBC.value)

            // 4. 把推荐列表保存到mongodb
            saveDataToMongoDB(userId, streamRecs)(mongoConfig, mongoClient)
          } finally {
            // 关闭连接
            jedis.close()
            mongoClient.close()
          }
      }
    }
  }

  /**
   * 从redis里获取最近num次评分
   */
  import scala.collection.JavaConversions._
  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {
    jedis.lrange("userId:" + userId.toString, 0, num)
      .map { item =>
        val attr = item.split("\\:")
        (attr(0).trim.toInt, attr(1).trim.toDouble)
      }
      .toArray
  }

  // 获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表
  def getTopSimProducts(num: Int,
                        productId: Int,
                        userId: Int,
                        simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                       (implicit mongoConfig: MongoConfig, mongoClient: MongoClient): Array[Int] = {
    // 从广播变量相似度矩阵中拿到当前商品的相似度列表
    val allSimProducts = simProducts(productId).toArray

    // 获得用户已经评分过的商品，过滤掉，排序输出
    val ratingCollection = mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    val ratingExist = ratingCollection.find(MongoDBObject("userId" -> userId))
      .toArray
      .map { item => // 只需要productId
        item.get("productId").toString.toInt
      }
    // 从所有的相似商品中进行过滤
    allSimProducts.filter(x => !ratingExist.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(num)
      .map(x => x._1)
  }

  // 计算每个备选商品的推荐得分
  def computeProductScore(candidateProducts: Array[Int],
                          userRecentlyRatings: Array[(Int, Double)],
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {
    // 定义一个长度可变数组ArrayBuffer，用于保存每一个备选商品的基础得分，(productId, score)
    val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    // 定义两个map，用于保存每个商品的高分和低分的计数器，productId -> count
    val increMap = scala.collection.mutable.HashMap[Int, Int]()
    val decreMap = scala.collection.mutable.HashMap[Int, Int]()

    // 遍历每个备选商品，计算和已评分商品的相似度
    for (candidateProduct <- candidateProducts; userRecentlyRating <- userRecentlyRatings) {
      // 从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
      val simScore = getProductsSimScore(candidateProduct, userRecentlyRating._1, simProducts)
      if (simScore > 0.4) {
        // 按照公式进行加权计算，得到基础评分
        scores += ((candidateProduct, simScore * userRecentlyRating._2))
        if (userRecentlyRating._2 > 3) {
          increMap(candidateProduct) = increMap.getOrDefault(candidateProduct, 0) + 1
        } else {
          decreMap(candidateProduct) = decreMap.getOrDefault(candidateProduct, 0) + 1
        }
      }
    }

    // 根据公式计算所有的推荐优先级，首先以productId做groupby
    scores.groupBy(_._1).map {
        case (productId, scoreList) =>
          (productId, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(productId, 1)) - log(decreMap.getOrDefault(productId, 1)))
      }
      // 返回推荐列表，按照得分排序
      .toArray
      .sortWith(_._2 > _._2)
  }

  def getProductsSimScore(product1: Int, product2: Int,
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    simProducts.get(product1) match { // 模式匹配
      case Some(sims) => sims.get(product2) match { // 二次模式匹配
        case Some(score) => score // 匹配到了直接返回score
        case None => 0.0 // 没有匹配到直接返回0.0
      }
      case None => 0.0
    }
  }

  // 自定义log函数，以N为底，实现了换底公式
  def log(m: Int): Double = {
    val N = 10
    math.log(m) / math.log(N)
  }

  // 写入mongodb
  def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig, mongoClient: MongoClient): Unit = {
    val streamRecsCollection = mongoClient(mongoConfig.db)(STREAM_RECS)
    // 按照userId查询并更新
    streamRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
    streamRecsCollection.insert(MongoDBObject("userId" -> userId,
      "recs" -> streamRecs.map(x => MongoDBObject("productId" -> x._1, "score" -> x._2))))
  }
}

case class MongoConfig(uri: String, db: String)